#include <zlib.h>
#include <set>

#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>

#include <log4cxx/xml/domconfigurator.h>

#include "SFSClient.h"

#define DS_STATUS_OK 0

using std::set;

LoggerPtr SFSClient::_logger = Logger::getLogger("SFSClient");

int SFSClient::init_logger(const char * log_conf_file) {
        if (NULL == log_conf_file) return -1;
        setlocale(LC_ALL, "zh_CN.GBK");
        _log_conf_file = log_conf_file;
        xml::DOMConfigurator::configure(_log_conf_file.c_str());
	return 1;
}

int SFSClient::init(const char * log_conf_file, const char * cs_ip, unsigned short cs_port, int cs_conn_pool_size, \
                        int cs_socket_conn_timeout, int cs_socket_send_timeout, int cs_socket_recv_timeout, \
                        int cs_socket_max_response_size) {
	if (NULL == log_conf_file || NULL == cs_ip) return -1;
	int ret = init_logger(log_conf_file);
	if (ret < 0) return -2;
	ret = _csc.init(cs_ip, cs_port, cs_conn_pool_size, cs_socket_conn_timeout, cs_socket_send_timeout, \
				cs_socket_recv_timeout, cs_socket_max_response_size);
        if (ret < 0) {
                LOG4CXX_ERROR(_logger, __FUNCTION__ << ", cnf svr client init error, ret:" << ret);
                return -3;
        }
	ret = _nscm.init(&_csc);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", name svr client manager init error, ret:" << ret);
		return -4;
	}
	ret = _mdscg.init(&_csc);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", meta data svr client group init error, ret:" << ret);
		return -5;
	}
	ret = _dscg.init(&_csc);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", data svr client group init error, ret:" << ret);
		return -6;
	}
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", init ok!");
	return 1;
}

int SFSClient::query_is_file_exist(const char * file_path, int & exist_flag) {
	if (NULL == file_path) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	NameSvrClient * nsc = _nscm.get_name_svr_client();
	if (NULL == nsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not obtain name svr client, file_path:" << file_path);
		return -2;
	}
	int ret = nsc->query_is_file_exist(file_path, exist_flag);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", name svr client query_is_file_exist error, ret:" << ret << \
				", file_path:" << file_path);
		return -3;
	}
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, file_path:" << file_path << ", exist_flag:" << exist_flag);
	return 1;
}

int SFSClient::query_is_dir_exist(const char * dir_path, int & exist_flag) {
	if (NULL == dir_path) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	NameSvrClient * nsc = _nscm.get_name_svr_client();
	if (NULL == nsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not obtain name svr client, dir_path:" << dir_path);
		return -2;
	}
	int ret = nsc->query_is_dir_exist(dir_path, exist_flag);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", name svr client query_is_dir_exist error, ret:" << ret << \
				", dir_path:" << dir_path);
		return -3;
	}
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, dir_path:" << dir_path << ", exist_flag:" << exist_flag);
	return 1;
}

int SFSClient::create_file(const char * file_path) {
	if (NULL == file_path) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	NameSvrClient * nsc = _nscm.get_name_svr_client();
	if (NULL == nsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not obtain name svr client, file_path:" << file_path);
		return -2;
	}
	int new_mds_id;
	int ret = nsc->alloc_mds_4_new_file(new_mds_id);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", alloc_mds_4_new_file error, file_path:" << file_path << ", ret:" << ret);
		return -3;
	}
	int new_ds_id;
	int32_t new_dir_id;
	int64_t new_file_id;
	MetadataSvrClient * mdsc = _mdscg.get_meta_data_svr_client(new_mds_id);
	if (NULL == mdsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not get meta data svr client, file_path:" << file_path);
		return -4;
	}
	ret = mdsc->alloc_ds_4_new_file(new_ds_id, new_dir_id, new_file_id);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", alloc_ds_4_new_file error, file_path:" << file_path << ", ret:" << ret);
		return -5;
	}
	DataSvrClient * dsc = _dscg.get_data_svr_client(new_ds_id);
	if (NULL == dsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not get_data_svr_client, file_path:" << file_path << \
				", new_ds_id:" << new_ds_id);
		return -6;
	}
	ret = mdsc->create_md_file(new_dir_id, new_file_id);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", meta data svr create file error, file_path:" << file_path << \
				", new_dir_id:" << new_dir_id << ", new_file_id:" << new_file_id << ",mds_id:" << \
				new_mds_id << ", ret:" << ret);
		return -7;
	}
	ret = dsc->create_file(new_dir_id, new_file_id);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", data svr create file error, file_path:" << file_path << ", dir_id:" << \
				new_dir_id << ", new_file_id:" << new_file_id << ", ds_id:" << new_ds_id << ", ret:" << ret);
		return -8;
	}
	ret = mdsc->client_notify_client_data_info(new_dir_id, new_file_id, vector<uint32_t>());
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", meta data svr client_notify_client_data_info error,  file_path:" << \
				file_path << ", new_dir_id:" << new_dir_id << ", new_file_id:" << new_file_id << \
				", new_mds_id:" << new_mds_id << \
				", ret:" << ret);
		return -9;
	}
	ret = nsc->create_file(file_path, new_mds_id, new_dir_id, new_file_id);	
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", name svr create file error, file_path:" << file_path << \
				", new_mds_id:" << new_mds_id << ", new_dir_id:" << new_dir_id << ", new_file_id:" << \
				new_file_id << ", ret:" << ret);
		return -10;
	}
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, file_path:" << file_path << \
				", new_mds_id:" << new_mds_id << ", new_dir_id:" << new_dir_id << ", new_file_id:" << \
				new_file_id << ", ret:" << ret);
	
	return 1;
}

int SFSClient::read_file(const char * file_path, char * buf, const int64_t begin_pos, const int len) {
	if (NULL == file_path || NULL == buf || len < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	if (0 == len) return 1;
	NameSvrClient * nsc = _nscm.get_name_svr_client();
	if (NULL == nsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not obtain name svr client, file_path:" << file_path);
		return -2;
	}
	FILE_METADATA_INFO fmdi;
	int ret = nsc->get_file_metadata_info(file_path, fmdi);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", name svr get_file_metadata_info error, file_path:" << \
				file_path << ", ret:" << ret);
		return -3;
	}
	MetadataSvrClient * mdsc = _mdscg.get_meta_data_svr_client(fmdi.mds_id);
	if (NULL == mdsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not meta data svr client, file_path:" << file_path << \
				", mds_id:" << fmdi.mds_id);
		return -4;
	}
	MD_FILE_INFO mdfi;
	ret = mdsc->get_md_file_info(fmdi.dir_id, fmdi.file_id, mdfi);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", meta data svr get_md_file_info error, file_path:" << file_path << \
				", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
				", ret:" << ret);
		return -5;
	}
	int ds_id = assign_a_ds_4_read(mdfi);
	if (ds_id < 1) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not get a data svr id to read file:" << file_path << \
					", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << \
					fmdi.mds_id << ", ret:" << ret);
		return -6;
	}
	int64_t filesize = calc_file_size(mdfi);
	if (filesize < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not determine file size, file:" << file_path <<  \
                                        ", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << \
                                        fmdi.mds_id << ", ret:" << ret);
		return -7;
	}
	int actual_need_to_read_size = len;
	if (begin_pos + len > filesize) actual_need_to_read_size = filesize - begin_pos;
	DataSvrClient * dsc = _dscg.get_data_svr_client(ds_id);
	if (NULL == dsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not get_data_svr_client, file_path:" << file_path << \
					", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << \
					fmdi.mds_id << ", ds_id:" << ds_id << ", ret:" << ret);
		return -8;
	}
	ret = dsc->read_file(fmdi.dir_id, fmdi.file_id, begin_pos, actual_need_to_read_size, buf);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", data svr read_file error, file_path:" << file_path << ", dir_id:" << \
                                        fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
                                        ", ds_id:" << ds_id << ", actual_need_to_read_size:" << actual_need_to_read_size << \
					", ret:" << ret);
		return -9;
	}
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, file_path:" << file_path << ", dir_id:" << \
                                        fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
                                        ", ds_id:" << ds_id << ", actual_need_to_read_size:" << actual_need_to_read_size);
	return actual_need_to_read_size;
}
	
int SFSClient::assign_a_ds_4_read(MD_FILE_INFO info) {
	set<int> data_ok_ds_id;
	for(map<int, DATA_SVR_MD_RECORD>::const_iterator iter = info.data_svr_md_records.begin(); \
		info.data_svr_md_records.end() != iter; iter++) {
		if (is_crcs_equal(iter->second.crcs, info.client_md_record.crcs)) {
			data_ok_ds_id.insert(iter->first);
		}
	}
	int total_weight = 0;
	list<DS_CNF> l_ds, l_perfect_ds;
	int ret = _csc.get_all_ds_cnf(l_ds);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", cnf server get_all_ds_cnf error, ret:" << ret);
		return -1;
	}
	for(list<DS_CNF>::const_iterator iter = l_ds.begin(); l_ds.end() != iter; iter++) {
		if (data_ok_ds_id.find(iter->id) == data_ok_ds_id.end()) continue;
		if (DS_STATUS_OK != iter->status) continue;
		if (iter->read_weight < 1) continue;
		l_perfect_ds.push_back(*iter);
		total_weight += iter->read_weight;
	}
	if (total_weight < 1) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not found a ds for read, total weight < 1");
		return -2;
	}
	int r = rand() % total_weight;
	int t = 0;
	for(list<DS_CNF>::const_iterator iter = l_perfect_ds.begin(); l_perfect_ds.end() != iter; iter++) {
		t += iter->read_weight;
		if (t >= r) return iter->id;
	}
	LOG4CXX_ERROR(_logger, __FUNCTION__ << ", unknown error, program should not reach here");
	return -100;
}

bool SFSClient::is_crcs_equal(vector<uint32_t> crcs1, vector<uint32_t> crcs2) {
        if (crcs1.size() != crcs2.size()) return false;
        int i = 0;
        for (vector<uint32_t>::const_iterator iter = crcs1.begin(); crcs1.end() != iter; iter++) {
                if (*iter != crcs2[i]) return false; 
                i++;
        }
        return true;
}

int SFSClient::delete_file(const char * file_path) {
	if (NULL == file_path) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	NameSvrClient * nsc = _nscm.get_name_svr_client();
	if (NULL == nsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not obtain name svr client, file_path:" << file_path);
		return -2;
	}
	FILE_METADATA_INFO fmdi;
	int ret = nsc->get_file_metadata_info(file_path, fmdi);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", name svr get_file_metadata_info error, file_path:" << \
				file_path << ", ret:" << ret);
		return -3;
	}
	MetadataSvrClient * mdsc = _mdscg.get_meta_data_svr_client(fmdi.mds_id);
	if (NULL == mdsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not meta data svr client, file_path:" << file_path << \
				", mds_id:" << fmdi.mds_id);
		return -4;
	}
	MD_FILE_INFO mdfi;
	ret = mdsc->get_md_file_info(fmdi.dir_id, fmdi.file_id, mdfi);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", meta data svr get_md_file_info error, file_path:" << file_path << \
				", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
				", ret:" << ret);
		return -5;
	}
	ret = nsc->delete_file(file_path);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", name svr delete_file error, file_path:" << file_path << \
                                ", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
                                ", ret:" << ret);
		return -6;
	}
	for(map<int, DATA_SVR_MD_RECORD>::const_iterator iter = mdfi.data_svr_md_records.begin(); \
		mdfi.data_svr_md_records.end() != iter; iter++) {
		DataSvrClient * dsc = _dscg.get_data_svr_client(iter->first);
		ret = dsc->delete_file(fmdi.dir_id, fmdi.file_id);
		if (ret < 0) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ",  data svr delete_file error, file_path:" <<  file_path << \
					", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << \
					fmdi.mds_id << " ds id:" << iter->first << ", ret:" << ret);
		}
	}
	ret = mdsc->delete_md_file(fmdi.dir_id, fmdi.file_id);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", meta data svr delete_md_file error, file_path:" <<  file_path << \
                                ", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
                                ", ret:" << ret);
	}
	return 1;
}

int SFSClient::create_dir(const char * dir_path) {
	if (NULL == dir_path) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	NameSvrClient * nsc = _nscm.get_name_svr_client();
	if (NULL == nsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not obtain name svr client, file_path:" << dir_path);
		return -2;
	}

	int ret = nsc->create_dir(dir_path);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", name svr create dir error, dir_path:" << dir_path << ", ret:" << ret);
		return -2;
	}
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, dir_path:" << dir_path);
	return 1;
}

int SFSClient::delete_dir(const char * dir_path) {
	if (NULL == dir_path) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	NameSvrClient * nsc = _nscm.get_name_svr_client();
	if (NULL == nsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not obtain name svr client, file_path:" << dir_path);
		return -2;
	}

	int ret = nsc->delete_dir(dir_path);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", name svr delete dir error, dir_path:" << dir_path << ", ret:" << ret);
		return -2;
	}
	LOG4CXX_INFO(_logger, __FUNCTION__ << ", ok, dir_path:" << dir_path);
	return 1;
}

int SFSClient::write_file(const char * file_path, const char * buf, const int len, int64_t begin_pos, int write_mode) {
	LOG4CXX_DEBUG(_logger, __FUNCTION__ << ", begin, file:" << file_path << ", len:" << len << ", begin_pos:" << begin_pos << \
				", write_mode:" << write_mode);
	if (NULL == file_path || NULL == buf || len < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	if (0 == len) return 1;
	NameSvrClient * nsc = _nscm.get_name_svr_client();
	if (NULL == nsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not obtain name svr client, file_path:" << file_path);
		return -2;
	}
	FILE_METADATA_INFO fmdi;
	int ret = nsc->get_file_metadata_info(file_path, fmdi);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", name svr get_file_metadata_info error, file_path:" << \
				file_path << ", ret:" << ret);
		return -3;
	}
	MetadataSvrClient * mdsc = _mdscg.get_meta_data_svr_client(fmdi.mds_id);
	if (NULL == mdsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not meta data svr client, file_path:" << file_path << \
				", mds_id:" << fmdi.mds_id);
		return -4;
	}
	MD_FILE_INFO mdfi;
	ret = mdsc->get_md_file_info(fmdi.dir_id, fmdi.file_id, mdfi);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", meta data svr get_md_file_info error, file_path:" << file_path << \
				", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
				", ret:" << ret);
		return -5;
	}
	int ds_id = alloc_a_ds_4_write(mdfi);
	if (ds_id < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", alloc_a_ds_4_write error, file_path:" << file_path << \
                                ", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
                                ", ret:" << ds_id);
		return -6;
	}
	vector<uint32_t> updated_client_crcs;
	ret = calc_updated_client_crcs(fmdi.mds_id, mdfi, fmdi.dir_id, fmdi.file_id, buf, len, begin_pos, write_mode, \
					updated_client_crcs);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", calc_updated_client_crcs error, file_path:" << file_path << \
				", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
				", ret:" << ret);
		return -7;
	}
	DataSvrClient * dsc = _dscg.get_data_svr_client(ds_id);
	if (NULL == dsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", get_data_svr_client error, file_path:" << file_path << \
				", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
				", ds_id :" << ds_id);
		return -8;
	}
	ret = dsc->write_file(fmdi.dir_id, fmdi.file_id, begin_pos, len, buf, write_mode);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", data svr write_file error, file_path:" << file_path << \
                                ", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
                                ", ds_id :" << ds_id << ", ret:" << ret);
		return -9;
	}
	ret = mdsc->client_notify_client_data_info(fmdi.dir_id, fmdi.file_id, updated_client_crcs);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", meta data svr client_notify_client_data_info error, file_path:" << \
				file_path << ", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << \
				fmdi.mds_id << ", ds_id :" << ds_id << ", ret:" << ret);
		return -10;
	}
	LOG4CXX_INFO(_logger, __FUNCTION__ << " ok, file_path:" << file_path << ", dir_id:" << fmdi.dir_id << ", file_id:" << \
				fmdi.file_id << ", mds_id:" << fmdi.mds_id << ", ds_id :" << ds_id << ", ret:" << ret);
	return 2;
}

int SFSClient::alloc_a_ds_4_write(MD_FILE_INFO info) {
	set<int> data_ok_ds_id;
	for(map<int, DATA_SVR_MD_RECORD>::const_iterator iter = info.data_svr_md_records.begin(); \
		info.data_svr_md_records.end() != iter; iter++) {
		if (is_crcs_equal(iter->second.crcs, info.client_md_record.crcs)) {
			data_ok_ds_id.insert(iter->first);
		}
	}
	int total_weight = 0;
	list<DS_CNF> l_ds, l_perfect_ds;
	int ret = _csc.get_all_ds_cnf(l_ds);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", cnf server get_all_ds_cnf error, ret:" << ret);
		return -1;
	}
	for(list<DS_CNF>::const_iterator iter = l_ds.begin(); l_ds.end() != iter; iter++) {
		if (data_ok_ds_id.find(iter->id) == data_ok_ds_id.end()) continue;
		if (DS_STATUS_OK != iter->status) continue;
		if (iter->write_weight < 1) continue;
		l_perfect_ds.push_back(*iter);
		total_weight += iter->write_weight;
	}
	if (total_weight < 1) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not found a ds for write, total weight < 1");
		return -2;
	}
	int r = rand() % total_weight;
	int t = 0;
	for(list<DS_CNF>::const_iterator iter = l_perfect_ds.begin(); l_perfect_ds.end() != iter; iter++) {
		t += iter->write_weight;
		if (t >= r) return iter->id;
	}
	LOG4CXX_ERROR(_logger, __FUNCTION__ << ", unknown error, program should not reach here");
	return -100;
}

int SFSClient::calc_updated_client_crcs(int mds_id, MD_FILE_INFO info, int32_t dir_id, int64_t file_id, \
		const char * buf, int len, int64_t begin_pos, int write_mode, \
		vector<uint32_t> & new_client_crcs) {
	if (mds_id < 1 || dir_id < 0 || file_id < 0 || NULL == buf || len < 1) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	int64_t read_begin_pos1 = -1;
	int64_t read_begin_pos2 = -1;
	int32_t read_len1 = -1;
	int32_t read_len2 = -1;
	int64_t file_size = -1;
	vector<uint32_t> old_client_crcs;
	old_client_crcs = info.client_md_record.crcs;
	for(map<int, DATA_SVR_MD_RECORD>::const_iterator iter = info.data_svr_md_records.begin(); \
			info.data_svr_md_records.end() != iter; iter++) {
		if (is_crcs_equal(iter->second.crcs, info.client_md_record.crcs)) {
			file_size = iter->second.file_size;
			break;
		}
	}
	if (-1 == file_size) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not determine file size, mds_id:" << mds_id << ", dir_id:" << \
				dir_id << ", file_id:" << file_id << ", write_len:" << len << ", begin_pos:" << begin_pos << \
				", write_mode:" << write_mode);
		return -2;
	}
	int ds_id = assign_a_ds_4_read(info);
	if (ds_id < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", assign_a_ds_for_read error, mds_id:" << mds_id << ", dir_id:" << \
				dir_id << ", file_id:" << file_id << ", write_len:" << len << ", begin_pos:" << begin_pos << \
				", write_mode:" << write_mode << ", ret:" << ds_id);
		return -3;
	}
	DataSvrClient * dsc = _dscg.get_data_svr_client(ds_id);
	if (NULL == dsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not obtain data svr client, mds_id:" << mds_id << ", dir_id:" << \
				dir_id << ", file_id:" << file_id << ", write_len:" << len << ", begin_pos:" << begin_pos << \
				", write_mode:" << write_mode << ", ds_id:" << ds_id);
		return -4;
	}

	if (DS_WRITE_MODE_APPEND == write_mode) {
		read_begin_pos1 = file_size - file_size % DS_CRC_CHUNK_SIZE;
		read_len1 = file_size % DS_CRC_CHUNK_SIZE;
		if (read_len1 > 0) {
			int begin_update_crc_index = read_begin_pos1 / DS_CRC_CHUNK_SIZE;
			char * tb = new char[read_len1 + len];
			int ret = dsc->read_file(dir_id, file_id, read_begin_pos1, read_len1, tb);
			if (ret < 0) {
				delete[] tb;
				LOG4CXX_ERROR(_logger, __FUNCTION__ << ", data svr read_file error, mds_id:" << mds_id << \
						", dir_id:" << dir_id << ", file_id:" << file_id << ", write_len:" << len << \
						", begin_pos:" << begin_pos << ", write_mode:" << write_mode << ", ds_id:" << \
						ds_id);
				return -5;
			}
			memcpy(tb + read_len1, buf, len);
			vector<uint32_t> t_crcs;
			for (int i = 0; i < read_len1 + len; i += DS_CRC_CHUNK_SIZE) {
				int crc_len = DS_CRC_CHUNK_SIZE;
				int left = read_len1 + len - i;
				if (left < DS_CRC_CHUNK_SIZE) crc_len = left;
				uint32_t crc_init = crc32(0L, NULL, 0);
				uint32_t crc = crc32(crc_init, (Bytef *)&tb[i], crc_len);
				t_crcs.push_back(crc);
			}
			for (int i =0; i < (begin_update_crc_index + t_crcs.size()); i++) {
				if (i < begin_update_crc_index) new_client_crcs.push_back(old_client_crcs[i]);
				else new_client_crcs.push_back(t_crcs[i - begin_update_crc_index]);
			}
			delete[] tb;
			return 1;
		}
		else {
			for (int i = 0; i < len; i += DS_CRC_CHUNK_SIZE) {
				int crc_len = DS_CRC_CHUNK_SIZE;
				int left = len - i;
				if (left < DS_CRC_CHUNK_SIZE) crc_len = left;
				uint32_t crc_init = crc32(0L, NULL, 0);
				uint32_t crc = crc32(crc_init, (Bytef *)&buf[i], crc_len);
				new_client_crcs.push_back(crc);
			}
			return 1;
		}
	}
	else {
		if (begin_pos > file_size) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", detected try to write a hole in the file, mds_id:" << mds_id << \
					", dir_id:" << dir_id << ", file_id:" << file_id << ", write_len:" << len << \
					", begin_pos:" << begin_pos << ", write_mode:" << write_mode << ", file_size:" << \
					file_size);
			return -6;
		}
		read_begin_pos1 = begin_pos -  begin_pos % DS_CRC_CHUNK_SIZE;
		read_len1 = begin_pos % DS_CRC_CHUNK_SIZE;
		int begin_update_crc_index = read_begin_pos1 / DS_CRC_CHUNK_SIZE;
		read_begin_pos2 = begin_pos + len;
		if (read_begin_pos2 > file_size || 0 == read_begin_pos2 % DS_CRC_CHUNK_SIZE) {
			read_len2 = 0;
		}
		else {
			read_len2 = DS_CRC_CHUNK_SIZE - read_begin_pos2 % DS_CRC_CHUNK_SIZE;
			if (read_begin_pos2 + read_len2 > file_size) {
				read_len2 = file_size - read_begin_pos2;
			}
		}
		char * tb = new char[read_len1 + read_len2 + len];
		if (read_len1 > 0) {
			int ret = dsc->read_file(dir_id, file_id, read_begin_pos1, read_len1, tb);
			if (ret < 0) {
				delete[] tb;
				LOG4CXX_ERROR(_logger, __FUNCTION__ << ", data svr read_file error, mds_id:" << mds_id << \
						", dir_id:" << dir_id << ", file_id:" << file_id << ", write_len:" << len << \
						", begin_pos:" << begin_pos << ", write_mode:" << write_mode << ", ds_id:" << \
						ds_id << ", ret:" << ret);
	
				return -7;
			}
		}
		memcpy(tb + read_len1, buf, len);
		if (read_len2 > 0) {
			int ret = dsc->read_file(dir_id, file_id, read_begin_pos2, read_len2, tb + read_len1 + len);
			if (ret < 0) {
				delete[] tb;
				LOG4CXX_ERROR(_logger, __FUNCTION__ << ", data svr read_file error, mds_id:" << mds_id << \
						", dir_id:" << dir_id << ", file_id:" << file_id << ", write_len:" << len << \
						", begin_pos:" << begin_pos << ", write_mode:" << write_mode << ", ds_id:" << \
						ds_id << ", ret:" << ret);
				return -8;
			}
		}
		vector<uint32_t> t_crcs;
		for (int i = 0; i < read_len1 + len + read_len2; i += DS_CRC_CHUNK_SIZE) {
			int crc_len = DS_CRC_CHUNK_SIZE;
			int left = read_len1 + len + read_len2 - i;
			if (left < DS_CRC_CHUNK_SIZE) crc_len = left;
			uint32_t crc_init = crc32(0L, NULL, 0);
			uint32_t crc = crc32(crc_init, (Bytef *)&tb[i], crc_len);
			t_crcs.push_back(crc);
		}
		delete[] tb;
		int end_update_crc_index = begin_update_crc_index + t_crcs.size() - 1;
		for (int i = 0; (i <= end_update_crc_index || i < old_client_crcs.size()); i++) {
			if (i < begin_update_crc_index || i > end_update_crc_index) new_client_crcs.push_back(old_client_crcs[i]);
			else new_client_crcs.push_back(t_crcs[i - begin_update_crc_index]);
		}
		return 1;
	}
}

int64_t SFSClient::get_file_size(const char * file_path) {
	if (NULL == file_path) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	NameSvrClient * nsc = _nscm.get_name_svr_client();
	if (NULL == nsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not obtain name svr client, file_path:" << file_path);
		return -2;
	}
	FILE_METADATA_INFO fmdi;	
	int ret = nsc->get_file_metadata_info(file_path, fmdi);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", name svr get_file_metadata_info error, file_path:" << \
				file_path << ", ret:" << ret);
		return -3;
	}
	MetadataSvrClient * mdsc = _mdscg.get_meta_data_svr_client(fmdi.mds_id);
	if (NULL == mdsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not meta data svr client, file_path:" << file_path << \
				", mds_id:" << fmdi.mds_id);
		return -4;
	}
	MD_FILE_INFO mdfi;
	ret = mdsc->get_md_file_info(fmdi.dir_id, fmdi.file_id, mdfi);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", meta data svr get_md_file_info error, file_path:" << file_path << \
				", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
				", ret:" << ret);
		return -5;
	}
	for(map<int, DATA_SVR_MD_RECORD>::const_iterator iter = mdfi.data_svr_md_records.begin(); \
			mdfi.data_svr_md_records.end() != iter; iter++) {
		if (is_crcs_equal(iter->second.crcs, mdfi.client_md_record.crcs)) {
			return iter->second.file_size;
		}
	}
	LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not find a valid replica for file:" << file_path);
	return -1;
}

int SFSClient::read_dir(const char * dir_path, list<string> & entries) {
	if (NULL == dir_path) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	NameSvrClient * nsc = _nscm.get_name_svr_client();
	if (NULL == nsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not obtain name svr client, file_path:" << dir_path);
		return -2;
	}
	int ret = nsc->read_dir(dir_path, entries);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", name svr client read_dir fail, ret:" << ret << ", dir_path:" << \
				dir_path);
		return -3;
	}
	return 1;
}

int SFSClient::get_file_stat(const char * file_path, struct stat & s) {
	if (NULL == file_path) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	NameSvrClient * nsc = _nscm.get_name_svr_client();
	if (NULL == nsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not obtain name svr client, file_path:" << file_path);
		return -2;
	}
	FILE_METADATA_INFO fmdi;
	int ret = nsc->get_file_metadata_info(file_path, fmdi);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", name svr get_file_metadata_info error, file_path:" << \
				file_path << ", ret:" << ret);
		return -3;
	}
	int len  = sizeof(struct stat);
	bzero(&s, len);
	s.st_nlink = fmdi.num_hard_link;
	if (SFS_FILE_TYPE_DIR == fmdi.file_type) {
		s.st_mode = S_IFDIR | 0777;
		s.st_size = 4096;
	}
	else if (SFS_FILE_TYPE_REG_FILE == fmdi.file_type) {
		s.st_mode =  S_IFREG | 0666;
		MetadataSvrClient * mdsc = _mdscg.get_meta_data_svr_client(fmdi.mds_id);
		if (NULL == mdsc) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not meta data svr client, file_path:" << file_path << \
					", mds_id:" << fmdi.mds_id);
			return -4;
		}
		MD_FILE_INFO mdfi;
		ret = mdsc->get_md_file_info(fmdi.dir_id, fmdi.file_id, mdfi);
		if (ret < 0) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", meta data svr get_md_file_info error, file_path:" << \
						file_path << ", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << \
						", mds_id:" << fmdi.mds_id << ", ret:" << ret);
			return -5;
		}
		s.st_size = calc_file_size(mdfi);
		if (s.st_size < 0) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not determine file size, file:" << file_path);
			return -6;
		}
	}
	return 1;
}

int64_t SFSClient::calc_file_size(MD_FILE_INFO mdfi) {
	map<int, DATA_SVR_MD_RECORD>::const_iterator iter;
	for(iter = mdfi.data_svr_md_records.begin(); mdfi.data_svr_md_records.end() != iter; iter++) {
		if (is_crcs_equal(iter->second.crcs, mdfi.client_md_record.crcs)) {
			return iter->second.file_size;
			
		}
	}
	LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not determine file size");
	return -1;
}

int SFSClient::truncate(const char * file_path, int64_t size) {
	if (NULL == file_path || size < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args");
		return -1;
	}
	NameSvrClient * nsc = _nscm.get_name_svr_client();
	if (NULL == nsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not obtain name svr client, file_path:" << file_path);
		return -2;
	}
	FILE_METADATA_INFO fmdi;
	int ret = nsc->get_file_metadata_info(file_path, fmdi);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", name svr get_file_metadata_info error, file_path:" << \
				file_path << ", ret:" << ret);
		return -3;
	}
	MetadataSvrClient * mdsc = _mdscg.get_meta_data_svr_client(fmdi.mds_id);
	if (NULL == mdsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not meta data svr client, file_path:" << file_path << \
				", mds_id:" << fmdi.mds_id);
		return -4;
	}
	MD_FILE_INFO mdfi;
	ret = mdsc->get_md_file_info(fmdi.dir_id, fmdi.file_id, mdfi);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", meta data svr get_md_file_info error, file_path:" << file_path << \
				", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
				", ret:" << ret);
		return -5;
	}
	int64_t old_file_size(-1);
	for(map<int, DATA_SVR_MD_RECORD>::const_iterator iter = mdfi.data_svr_md_records.begin(); \
			mdfi.data_svr_md_records.end() != iter; iter++) {
		if (is_crcs_equal(iter->second.crcs, mdfi.client_md_record.crcs)) {
			old_file_size = iter->second.file_size;
			break;
		}
	}
	if (-1 == old_file_size) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not determine file size, mds_id:" << fmdi.mds_id << ", dir_id:" << \
				fmdi.dir_id << ", file_id:" << fmdi.file_id << ", file:" << file_path);
		return -6;
	}

	if (size == old_file_size) {
		LOG4CXX_INFO(_logger, __FUNCTION__ << ", file size is the same, no need to truncate, file:" << file_path);
		return 1;
	}

	int ds_id = alloc_a_ds_4_write(mdfi);
	if (ds_id < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", alloc_a_ds_4_write error, file_path:" << file_path << \
                                ", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
                                ", ret:" << ds_id);
		return -7;
	}
	vector<uint32_t> updated_client_crcs;
	ret = calc_updated_client_crcs_4_truncate(fmdi.mds_id, mdfi, fmdi.dir_id, fmdi.file_id, size, updated_client_crcs);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", calc_updated_client_crcs_4_truncate error, file_path:" << file_path << \
				", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
				", ret:" << ret);
		return -8;
	}
	DataSvrClient * dsc = _dscg.get_data_svr_client(ds_id);
	if (NULL == dsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", get_data_svr_client error, file_path:" << file_path << \
				", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
				", ds_id :" << ds_id);
		return -9;
	}
	ret = dsc->truncate(fmdi.dir_id, fmdi.file_id, size);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", data svr truncate error, file_path:" << file_path << \
                                ", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << fmdi.mds_id << \
                                ", ds_id :" << ds_id << ", ret:" << ret);
		return -10;
	}
	ret = mdsc->client_notify_client_data_info(fmdi.dir_id, fmdi.file_id, updated_client_crcs);
	if (ret < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", meta data svr client_notify_client_data_info error, file_path:" << \
				file_path << ", dir_id:" << fmdi.dir_id << ", file_id:" << fmdi.file_id << ", mds_id:" << \
				fmdi.mds_id << ", ds_id :" << ds_id << ", ret:" << ret);
		return -11;
	}
	LOG4CXX_INFO(_logger, __FUNCTION__ << " ok, file_path:" << file_path << ", dir_id:" << fmdi.dir_id << ", file_id:" << \
				fmdi.file_id << ", mds_id:" << fmdi.mds_id << ", ds_id :" << ds_id << ", ret:" << ret);
	return 2;

}

int SFSClient::calc_updated_client_crcs_4_truncate(int mds_id, MD_FILE_INFO info, int32_t dir_id, int64_t file_id, \
		int64_t size, vector<uint32_t> & new_client_crcs) {
	if (mds_id < 0 || dir_id < 0 || file_id < 0 || size < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", invalid args, mds_id:" << mds_id << ", dir_id:" << dir_id << \
				", file_id:" << file_id << ", size:" << size);
		return -1;
	}
	vector<uint32_t> old_client_crcs = info.client_md_record.crcs;
	int64_t old_file_size(-1);
	for(map<int, DATA_SVR_MD_RECORD>::const_iterator iter = info.data_svr_md_records.begin(); \
			info.data_svr_md_records.end() != iter; iter++) {
		if (is_crcs_equal(iter->second.crcs, info.client_md_record.crcs)) {
			old_file_size = iter->second.file_size;
			break;
		}
	}
	if (-1 == old_file_size) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not determine file size, mds_id:" << mds_id << ", dir_id:" << \
				dir_id << ", file_id:");
		return -2;
	}
	int ds_id = assign_a_ds_4_read(info);
	if (ds_id < 0) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", assign_a_ds_for_read error, mds_id:" << mds_id << ", dir_id:" << \
				dir_id << ", file_id:" << file_id << ", ret:" << ds_id);
		return -3;
	}
	DataSvrClient * dsc = _dscg.get_data_svr_client(ds_id);
	if (NULL == dsc) {
		LOG4CXX_ERROR(_logger, __FUNCTION__ << ", could not obtain data svr client, mds_id:" << mds_id << ", dir_id:" << \
				dir_id << ", file_id:" << file_id << ", ds_id:" << ds_id);
		return -4;
	}
	if (old_file_size > size) {
		int64_t t = (size /  DS_CRC_CHUNK_SIZE);
		int64_t read_begin_pos = t * DS_CRC_CHUNK_SIZE;
		int32_t read_len = size % DS_CRC_CHUNK_SIZE;
		int32_t end_chunk_crc;
		if (read_len > 0) {
			char * buf = new char[read_len];
			int ret = dsc->read_file(dir_id, file_id, read_begin_pos, read_len, buf);
			if (ret < 0) {
				LOG4CXX_ERROR(_logger, __FUNCTION__ << ", data svr read_file error,  dir_id:" << \
						dir_id << ", file_id:" << file_id << ", mds_id:" << mds_id << \
						", ds_id:" << ds_id << ", read_len:" << read_len<< ", ret:" << ret);
				delete[] buf;
				return -5;
			}
			uint32_t crc_init = crc32(0L, NULL, 0);
			end_chunk_crc = crc32(crc_init, (Bytef *)buf, read_len);
			delete[] buf;
		}
		int new_num_of_ds_trunk;
		if (0 == read_len) new_num_of_ds_trunk = (size /  DS_CRC_CHUNK_SIZE);
		else new_num_of_ds_trunk = (size /  DS_CRC_CHUNK_SIZE) + 1;
		vector<uint32_t>::const_iterator iter = old_client_crcs.begin();
		for (int i = 0; i < new_num_of_ds_trunk; i++, iter++) {
			if ((i == new_num_of_ds_trunk - 1) && 0 != read_len) {
				new_client_crcs.push_back(end_chunk_crc);
			}
			else new_client_crcs.push_back(*iter);
		}
		return 1;
	}
	// the following statements handle old_file_size < size
	int64_t t = (old_file_size /  DS_CRC_CHUNK_SIZE);
	int64_t read_begin_pos = t * DS_CRC_CHUNK_SIZE;
	int32_t read_len = old_file_size % DS_CRC_CHUNK_SIZE;
	int64_t buf_len = size - read_begin_pos;
	char * buf = new char[buf_len];
	bzero(buf, buf_len);
	if (read_len > 0) {
		int ret = dsc->read_file(dir_id, file_id, read_begin_pos, read_len, buf);
		if (ret < 0) {
			LOG4CXX_ERROR(_logger, __FUNCTION__ << ", data svr read_file error,  dir_id:" << \
					dir_id << ", file_id:" << file_id << ", mds_id:" << mds_id << \
					", ds_id:" << ds_id << ", read_len:" << read_len<< ", ret:" << ret);
			delete[] buf;
			return -6;
		}
	}
	vector<uint32_t> append_crcs;
	for (int i = 0; i < size - read_begin_pos; i+= DS_CRC_CHUNK_SIZE) {
		uint32_t crc_init = crc32(0L, NULL, 0);
		int64_t left = buf_len - i;
		int crc_len = DS_CRC_CHUNK_SIZE;
		if (left < DS_CRC_CHUNK_SIZE) crc_len = left;
		uint32_t the_crc = crc32(crc_init, (Bytef *)&(buf[i]), crc_len);
		append_crcs.push_back(the_crc);
	}
	delete[] buf;
	int begin_update_crc_index;
	vector<uint32_t>::const_iterator iter1 = old_client_crcs.begin();
	int old_crc_size = old_client_crcs.size();
	vector<uint32_t>::const_iterator iter2 = append_crcs.begin();
	int i = 0;
	while (append_crcs.end() != iter2) {
		if (old_client_crcs.end() != iter1) {
			if (old_crc_size - 1 != i || 0 == read_len) {
				new_client_crcs.push_back(*iter1);
			}
			iter1++;
			i++;
		}
		else {
			new_client_crcs.push_back(*iter2);
			iter2++;
		}
	}
	return 2;
}
